package com.bw.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import com.bw.utils.MyClickHouseUtil;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class ShopMetricsCalculation {

    // 页面日志数据模型
    public static class PageLog implements Serializable {
        private String uid;
        private String pageId;
        private long duringTime;
        private String lastPageId;
        private String isNew;
        private long ts;  // 事件时间戳

        // 构造函数、getter和setter
        public PageLog() {}

        public PageLog(String uid, String pageId, long duringTime, String lastPageId, String isNew, long ts) {
            this.uid = uid;
            this.pageId = pageId;
            this.duringTime = duringTime;
            this.lastPageId = lastPageId;
            this.isNew = isNew;
            this.ts = ts;
        }

        // Getters and Setters
        public String getUid() { return uid; }
        public void setUid(String uid) { this.uid = uid; }
        public String getPageId() { return pageId; }
        public void setPageId(String pageId) { this.pageId = pageId; }
        public long getDuringTime() { return duringTime; }
        public void setDuringTime(long duringTime) { this.duringTime = duringTime; }
        public String getLastPageId() { return lastPageId; }
        public void setLastPageId(String lastPageId) { this.lastPageId = lastPageId; }
        public String getIsNew() { return isNew; }
        public void setIsNew(String isNew) { this.isNew = isNew; }
        public long getTs() { return ts; }
        public void setTs(long ts) { this.ts = ts; }
    }

    // 计算结果实体类（与ClickHouse表结构对应）
    public static class ShopMetrics {
        private String windowStart;  // 窗口开始时间
        private String windowEnd;    // 窗口结束时间
        private String processTime;  // 处理时间
        private long visitorCount;   // 访客数
        private long pageViewCount;  // 浏览量
        private double bounceRate;   // 跳失率
        private double avgStayTime;  // 平均停留时长
        private long newVisitorCount;// 新访客数
        private long oldVisitorCount;// 老访客数

        // 构造函数、getter和setter
        public ShopMetrics() {}

        public ShopMetrics(String windowStart, String windowEnd, String processTime,
                          long visitorCount, long pageViewCount, double bounceRate,
                          double avgStayTime, long newVisitorCount, long oldVisitorCount) {
            this.windowStart = windowStart;
            this.windowEnd = windowEnd;
            this.processTime = processTime;
            this.visitorCount = visitorCount;
            this.pageViewCount = pageViewCount;
            this.bounceRate = bounceRate;
            this.avgStayTime = avgStayTime;
            this.newVisitorCount = newVisitorCount;
            this.oldVisitorCount = oldVisitorCount;
        }

        // Getters and Setters
        public String getWindowStart() { return windowStart; }
        public void setWindowStart(String windowStart) { this.windowStart = windowStart; }
        public String getWindowEnd() { return windowEnd; }
        public void setWindowEnd(String windowEnd) { this.windowEnd = windowEnd; }
        public String getProcessTime() { return processTime; }
        public void setProcessTime(String processTime) { this.processTime = processTime; }
        public long getVisitorCount() { return visitorCount; }
        public void setVisitorCount(long visitorCount) { this.visitorCount = visitorCount; }
        public long getPageViewCount() { return pageViewCount; }
        public void setPageViewCount(long pageViewCount) { this.pageViewCount = pageViewCount; }
        public double getBounceRate() { return bounceRate; }
        public void setBounceRate(double bounceRate) { this.bounceRate = bounceRate; }
        public double getAvgStayTime() { return avgStayTime; }
        public void setAvgStayTime(double avgStayTime) { this.avgStayTime = avgStayTime; }
        public long getNewVisitorCount() { return newVisitorCount; }
        public void setNewVisitorCount(long newVisitorCount) { this.newVisitorCount = newVisitorCount; }
        public long getOldVisitorCount() { return oldVisitorCount; }
        public void setOldVisitorCount(long oldVisitorCount) { this.oldVisitorCount = oldVisitorCount; }
    }

    // 聚合状态类
    public static class MetricsAgg implements Serializable{
        private long pageViewCount;    // 浏览量
        private long bounceCount;      // 跳失次数
        private long totalStayTime;    // 总停留时间
        private Set<String> newVisitors = new HashSet<>();  // 新访客去重
        private Set<String> oldVisitors = new HashSet<>();  // 老访客去重

        public MetricsAgg() {}
    }

    public static void main(String[] args) throws Exception {
        // 1. 初始化执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  // 测试阶段设置为1，生产环境根据集群调整



        // 配置检查点 - 新增代码
        env.enableCheckpointing(5000);  // 每5秒进行一次检查点
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);

        // 2. 配置Kafka消费者
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "hadoop102:9092");
        kafkaProps.setProperty("group.id", "shop_metrics_group");

        // 3. 读取Kafka数据
        DataStream<String> kafkaSource = env.addSource(
                new FlinkKafkaConsumer<>("dwd_traffic_page_log", new SimpleStringSchema(), kafkaProps)
                        .setStartFromEarliest()
        );

        // 4. 解析JSON为PageLog对象并提取事件时间
        // 解析JSON数据时添加校验
        DataStream<PageLog> pageLogStream = kafkaSource
                .map(jsonStr -> {
                    try {
                        JSONObject json = JSON.parseObject(jsonStr);
                        // 校验必要字段存在
                        if (!json.containsKey("common") || !json.containsKey("page") || !json.containsKey("ts")) {
                            System.err.println("缺少必要字段: " + jsonStr);
                            return null;
                        }

                        JSONObject common = json.getJSONObject("common");
                        JSONObject page = json.getJSONObject("page");

                        // 校验时间戳有效性
                        long ts = json.getLongValue("ts");
                        if (ts <= 0) {
                            System.err.println("无效的时间戳: " + jsonStr);
                            return null;
                        }

                        return new PageLog(
                                common.getString("uid"),
                                page.getString("page_id"),
                                page.getLongValue("during_time"),
                                page.getString("last_page_id"),
                                common.getString("is_new"),
                                ts
                        );
                    } catch (Exception e) {
                        System.err.println("解析失败: " + jsonStr + ", 错误: " + e.getMessage());
                        e.printStackTrace();
                        return null;
                    }
                })
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<PageLog>() {
                    private final long maxOutOfOrderness = 5000L;  // 允许5秒乱序
                    private long currentMaxTimestamp = 0;

                    @Overrideg
                    public Watermark getCurrentWatermark() {
                        // 确保currentMaxTimestamp有初始值，避免生成负的水印
                        return new Watermark(currentMaxTimestamp > maxOutOfOrderness ?
                                currentMaxTimestamp - maxOutOfOrderness : 0);
                    }

                    @Override
                    public long extractTimestamp(PageLog element, long previousElementTimestamp) {
                        long timestamp = element.getTs();
                        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                        return timestamp;
                    }
                });

        // 5. 按店铺分组（假设固定店铺，实际按业务字段分组）+ 窗口计算
        DataStream<ShopMetrics> metricsStream = pageLogStream
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))  // 10分钟滚动窗口
                .aggregate(
                        new AggregateFunction<PageLog, MetricsAgg, MetricsAgg>() {
                            @Override
                            public MetricsAgg createAccumulator() {
                                return new MetricsAgg();
                            }

                            @Override
                            public MetricsAgg add(PageLog pageLog, MetricsAgg accumulator) {
                                // 原有实现逻辑保持不变
                                accumulator.pageViewCount++;

                                // 判断是否为跳失行为（停留时间短且没有后续页面）
                                if (pageLog.getDuringTime() < 1000 && pageLog.getLastPageId() == null) {
                                    accumulator.bounceCount++;
                                }

                                // 累加停留时间
                                accumulator.totalStayTime += pageLog.getDuringTime();

                                // 根据isNew标记统计新老访客
                                if ("1".equals(pageLog.getIsNew())) {
                                    accumulator.newVisitors.add(pageLog.getUid());
                                } else {
                                    accumulator.oldVisitors.add(pageLog.getUid());
                                }

                                return accumulator;
                            }

                            @Override
                            public MetricsAgg getResult(MetricsAgg accumulator) {
                                return accumulator;
                            }

                            @Override
                            public MetricsAgg merge(MetricsAgg a, MetricsAgg b) {
                                // 合并两个累加器的统计数据
                                MetricsAgg result = new MetricsAgg();

                                // 合并浏览量
                                result.pageViewCount = a.pageViewCount + b.pageViewCount;

                                // 合并跳失次数
                                result.bounceCount = a.bounceCount + b.bounceCount;

                                // 合并总停留时间
                                result.totalStayTime = a.totalStayTime + b.totalStayTime;

                                // 合并新访客集合
                                result.newVisitors.addAll(a.newVisitors);
                                result.newVisitors.addAll(b.newVisitors);

                                // 合并老访客集合
                                result.oldVisitors.addAll(a.oldVisitors);
                                result.oldVisitors.addAll(b.oldVisitors);

                                return result;
                            }
                        },
                        new ProcessAllWindowFunction<MetricsAgg, ShopMetrics, TimeWindow>() {
                            private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                            @Override
                            public void process(Context context, Iterable<MetricsAgg> elements, Collector<ShopMetrics> out) {
                                MetricsAgg agg = elements.iterator().next();
                                TimeWindow window = context.window();

                                // 计算指标
                                long visitorCount = agg.newVisitors.size() + agg.oldVisitors.size();
                                double bounceRate = agg.pageViewCount > 0 ?
                                        (double) agg.bounceCount / agg.pageViewCount : 0;
                                double avgStayTime = agg.pageViewCount > 0 ?
                                        (double) agg.totalStayTime / agg.pageViewCount : 0;

                                // 构建结果对象
                                ShopMetrics metrics = new ShopMetrics(
                                        sdf.format(new Date(window.getStart())),
                                        sdf.format(new Date(window.getEnd())),
                                        sdf.format(new Date()),  // 处理时间（当前系统时间）
                                        visitorCount,
                                        agg.pageViewCount,
                                        bounceRate,
                                        avgStayTime,
                                        agg.newVisitors.size(),
                                        agg.oldVisitors.size()
                                );
                                out.collect(metrics);
                            }
                        }
                );

        // 6. 调用ClickHouse工具类写入数据
        String insertSql = "INSERT INTO gmall.dws_shop_metrics (" +
                "window_start, window_end, process_time, " +
                "visitor_count, page_view_count, bounce_rate, " +
                "avg_stay_time, new_visitor_count, old_visitor_count" +
                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";

        SinkFunction<ShopMetrics> clickhouseSink = MyClickHouseUtil.getSinkFunction(insertSql);
        metricsStream.addSink(clickhouseSink);

        // 7. 执行任务
        env.execute("Shop Metrics Calculation");
    }
}